package com.aotain.nyx.udpflood;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.flink.BasicTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

public class UDPFloodDetect  implements FlatMapFunction<Tuple2<String,HashMap<String,BasicTuple>>, AbnStatisTuple>{

	/**
	 *
	 */
	    
	private static final long serialVersionUID = 6372424784680465563L;
	
	/**
	 * 固定源60s内累计包数目标|源|端口
	 */
	private HashMap<String,Long> fixedSourceMap = new HashMap<String,Long>();

	/**
	 * 伪造源60s内累计包  目标|端口
	 */
	private HashMap<String,Long> forgedSourceMap = new HashMap<String,Long>();
	

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();
	
//	/**
//	 * 上行包字节数
//	 */
//	private HashMap<String,Long> mapStreamoctets = new HashMap<String,Long>();
//	
//	/**
//	 * 上行包数
//	 */
//	private HashMap<String,Long> mapStreampacket = new HashMap<String,Long>();
	
	/**
	 * 持久-上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctetsMem = new HashMap<String,Long>();
	
	/**
	 * 持久-上行包数
	 */
	private HashMap<String,Long> mapStreampacketMem = new HashMap<String,Long>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();

	@Override
	public void flatMap(Tuple2<String, HashMap<String,BasicTuple>> value,
			Collector<AbnStatisTuple> out) throws Exception {
		
		String groupip = value.f0;
		HashMap<String,BasicTuple> tupleMap = value.f1;
		//value 包含一个目的IP，在1分钟粒度内，所有访问他的源IP的集合
		
		//计算IP的总包数
		long totalPack = 0l;
		long totalOct = 0l;
		for(Entry<String,BasicTuple> entry : tupleMap.entrySet()) {
			totalPack = totalPack + entry.getValue().getUpStreamPacket();
			totalOct = totalOct + entry.getValue().getUpStreamPacket();
		}
		
		long currenttime = System.currentTimeMillis();
		
		if(totalPack/60 >= 100000) {
			//满足伪造源门限执行伪造源流程
			String forgedKey = groupip + "ISFORGED" ;
			String destip = groupip.split("_",-1)[0];
			String destport = groupip.split("_",-1)[1];
			BasicTuple tupleforged = new BasicTuple();
			tupleforged.setDestIP(destip);
			tupleforged.setDestPort(destport);
			tupleforged.setSourceIP("ANY");
			tupleforged.setUpStreamOctets(totalOct);
			tupleforged.setUpStreamPacket(totalPack);
			
			if(attackLasttimeMap.containsKey(forgedKey )) {//如果存在就更新
				attackLasttimeMap.put(groupip, currenttime);
				
				AbnStatisTuple abnTuple = UpdateData(tupleforged, false);
				out.collect(abnTuple);
			}else {//不存在就新入数据
				attackStartMap.put(forgedKey, currenttime);
				attackLasttimeMap.put(forgedKey, currenttime);
				AbnStatisTuple abnTuple = InsertIntoData(tupleforged, false, currenttime);
				out.collect(abnTuple);
			}
			
		} else {
			
			//-----------伪造源 START --------------
			String forgedKey = groupip;
			String destip = groupip.split("_",-1)[0];
			String destport = groupip.split("_",-1)[1];
			BasicTuple tupleforged = new BasicTuple();
			tupleforged.setDestIP(destip);
			tupleforged.setDestPort(destport);
			tupleforged.setSourceIP("ANY");
			tupleforged.setUpStreamOctets(totalOct);
			tupleforged.setUpStreamPacket(totalPack);
			
			if (attackLasttimeMap.containsKey(forgedKey)) {
				if(currenttime - attackLasttimeMap.get(forgedKey) > END_DURATION) {

					AbnStatisTuple abnTuple = EndIntoData(tupleforged, false);	
					out.collect(abnTuple);
					attackLasttimeMap.remove(forgedKey);
					attackStartMap.remove(forgedKey);
					mapStreamoctetsMem.remove(forgedKey);
					mapStreampacketMem.remove(forgedKey);
					mapHighFlow.remove(forgedKey);
				}
			}
			//-----------伪造源 END --------------
			
			//判断单个源IP访问，是否满足固定源UDP Flood攻击
			for(Entry<String,BasicTuple> entry : tupleMap.entrySet()) {
				BasicTuple tuple = entry.getValue();
				long entryValue = tuple.getUpStreamPacket();
				String dip = tuple.getDestIP();
				String mapKey = String.format("%s_%s_%s", dip, tuple.getSourceIP(), tuple.getDestPort());
				if(entryValue/60 >= 100000) {//如果每秒上行包总数大于10000则判断有攻击					
				    if(attackLasttimeMap.containsKey(mapKey)) {//如果存在就更新
						attackLasttimeMap.put(mapKey, currenttime);
						AbnStatisTuple abnTuple = UpdateData(tuple, true);
						out.collect(abnTuple);
						
					}else {//不存在就新入数据
						attackStartMap.put(mapKey, currenttime);
						attackLasttimeMap.put(mapKey, currenttime);
						
						AbnStatisTuple abnTuple = InsertIntoData(tuple, true, currenttime);
						out.collect(abnTuple);
					}
				} else {
					//如果没大于门限 且过期3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(mapKey)) {
						if(currenttime - attackLasttimeMap.get(mapKey) > END_DURATION) {
							AbnStatisTuple abnTuple = EndIntoData(tuple, true);	
							out.collect(abnTuple);
							attackLasttimeMap.remove(mapKey);
							attackStartMap.remove(mapKey);
							mapStreamoctetsMem.remove(mapKey);
							mapStreampacketMem.remove(mapKey);
							mapHighFlow.remove(mapKey);
						}
					}
				}
			
			}
		}
		
		
		ArrayList<String> removeItem = new ArrayList<String>();			
		for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
			String mapKey = entry.getKey();
				
			if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {					
				
				removeItem.add(mapKey);
				if(mapKey.endsWith("ISFORGED")) {	
					//伪造源
					String destip = groupip.split("_",-1)[0];
					String destport = groupip.split("_",-1)[1];
					BasicTuple tupleforged = new BasicTuple();
					tupleforged.setDestIP(destip);
					tupleforged.setDestPort(destport);
					tupleforged.setSourceIP("ANY");
					tupleforged.setUpStreamOctets(totalOct);
					tupleforged.setUpStreamPacket(totalPack);
					AbnStatisTuple abnTuple = EndIntoData(tupleforged, false);
					out.collect(abnTuple);
					
				} else {
					//固定源
					for(Entry<String,BasicTuple> entryTuple : tupleMap.entrySet()) {
						if(entryTuple.getKey().startsWith(mapKey)) {
							AbnStatisTuple abnTuple = EndIntoData(entryTuple.getValue(),true);
							out.collect(abnTuple);
						}
					}
					
					
				}					
//				attackLasttimeMap.remove(mapKey);
				attackStartMap.remove(mapKey);
				mapStreamoctetsMem.remove(mapKey);
				mapStreampacketMem.remove(mapKey);
			}
		}

		for(String key  : removeItem) {
			attackLasttimeMap.remove(key);
		}
	
		removeItem.clear();
		fixedSourceMap.clear();
		forgedSourceMap.clear();
	}
	
	
	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public AbnStatisTuple InsertIntoData(BasicTuple tuple,boolean isFixed, long time) {

		String destAreaName = "";
		String destGis = "";
		String sourceAreaName = "";
		String sourceGis = "";
		String sourceAreaCountry = "";
		String sourceAreaId = "";		
		String sourceProvinceName =  "";//如果省为空，精确到国家
		String sourceAreaCityId = "";
		String sourceAreaProvinceId = "";
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		String dport = tuple.getDestPort();
		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String idc = tuple.getIDC();
		String[] gisArray = gis.split("#");								
		destAreaName = gisArray[0];
		destGis = gisArray[1];
		sourceAreaName = gisArray[2];
		sourceGis = gisArray[3];
		sourceAreaCountry = gisArray[4];
		sourceAreaId = gisArray[5];		
		sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
		sourceAreaCityId = gisArray[7];
		sourceAreaProvinceId = gisArray[8];
		String desc = "";
		String rowKeyAbnormal = "";
		
		String mapKey = String.format("%s_%s_%s", dip, sip, dport);

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));
		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		String rowkey3 = String.format("%s_%s", accesstime, sip);
		
		long entryValue = tuple.getUpStreamPacket();
		
		if(isFixed) {
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, tuple.getUpStreamOctets()/(1024*60));
		} else {
			dip = tuple.getDestIP();
			sip = "伪造源";
			dport = tuple.getDestPort();
			idc = tuple.getIDC();
			sourceAreaName = "伪造源";
			//				sourceGis = "ANY";
			sourceAreaCountry = "伪造源";
			sourceAreaId = "伪造源";	
			sourceProvinceName = "伪造源"; //如果省为空，精确到国家
			sourceAreaCityId = "伪造源";
			sourceAreaProvinceId = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport +  "_7";
			desc = String.format("UDPFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]",entryValue, tuple.getUpStreamOctets()/(1024*60));
		}

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ROWKEY", rowKeyAbnormal));
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", dport));
		row.add(new DataColumn("ACCESSTIME", accesstime));
		row.add(new DataColumn("ABRNORMAL", "7"));
		row.add(new DataColumn("DESTIP", dip));							
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));							
		row.add(new DataColumn("ATTNUM", tuple.getUpStreamPacket()));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("UPSTREAMOCTETS", tuple.getUpStreamOctets()));	
		row.add(new DataColumn("UPSTREAMPACKET", tuple.getUpStreamPacket()));
		row.add(new DataColumn("ATTTYPE", "1"));
		row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));
		
		importtool.InsertRowData(tableName, row);

		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		//if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = tuple.getUpStreamOctets();
			nUpstreampacket = tuple.getUpStreamPacket();
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			row.add(new DataColumn("UPSTREAMOCTETS", nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET", nUpstreampacket));
			
			mapStreamoctetsMem.put(mapKey, nUpStreamOctets);
			mapStreampacketMem.put(mapKey, nUpstreampacket);
		}
		
		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);
		
		double maxValue =  nUpStreamOctets/(1024*60);
		mapHighFlow.put(mapKey, maxValue);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("UDPFLOOD");
		abnTuple.setAttackNum(1l);
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("1");
		abnTuple.setAttTime(accesstime);
		abnTuple.setAttackMax(maxValue);
		
		return abnTuple;

	}
	
	
	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	public AbnStatisTuple UpdateData(BasicTuple tuple,boolean isFixed) {

		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		String desc = "";
		String dport = tuple.getDestPort();
		String rowKeyAbnormal = "";
		String mapKey = String.format("%s_%s_%s", dip, sip, dport);
		
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(mapKey)));

		long entryValue = tuple.getUpStreamPacket();
		
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		//if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = tuple.getUpStreamOctets();
			nUpstreampacket = tuple.getUpStreamPacket();
			if(mapStreamoctetsMem.containsKey(mapKey))
				mapStreamoctetsMem.put(mapKey, mapStreamoctetsMem.get(mapKey) + nUpStreamOctets);
			if(mapStreampacketMem.containsKey(mapKey))
				mapStreampacketMem.put(mapKey, mapStreampacketMem.get(mapKey) + nUpstreampacket);
		}
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
	
		//if(mapStreamoctetsMem.size() > 0 && mapStreamoctetsMem.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(mapKey);
			nUpstreampacket = mapStreampacketMem.get(mapKey);
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
		}
		
		if(isFixed) {
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, nUpStreamOctets/(1024*60));			
		} else {
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_7";
			desc = String.format("UDPFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, nUpStreamOctets/(1024*60));
		}
		
		
		row.add(new DataColumn("UPSTREAMOCTETS", nUpStreamOctets));	
		row.add(new DataColumn("UPSTREAMPACKET", nUpstreampacket));
		row.add(new DataColumn("ATTNUM", nUpstreampacket));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("ENDTIME", Long.parseLong(accesstime)));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);


		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuple.getIDC();
		
		double maxValue = (double)nUpStreamOctets/(double)60/(double)1024 ;

		if(mapHighFlow.containsKey(mapKey))
		{
			double maxValueMem = mapHighFlow.get(mapKey);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(mapKey, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(mapKey, maxValue);
		}
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("UDPFLOOD");
		abnTuple.setUpStreamOctets(tuple.getUpStreamOctets());
		abnTuple.setUpStreamPacket(tuple.getUpStreamPacket());
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("1");
		abnTuple.setAttackMax(maxValue);
		abnTuple.setAttTime(accesstime);
		return abnTuple;

	}
	
	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: 攻击结束记录入库
	 * @param entryKey
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	public AbnStatisTuple EndIntoData(BasicTuple tuple,boolean isFixed) {
		
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		String desc = "";
		String dport = tuple.getDestPort();
		String rowKeyAbnormal = "";
		String mapKey = String.format("%s_%s_%s", dip, sip, dport);
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(mapKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);


		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		//if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(mapKey) + tuple.getUpStreamOctets();
			nUpstreampacket = mapStreampacketMem.get(mapKey) + tuple.getUpStreamPacket();
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
		}
		
		if(isFixed) {
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(mapKey)-attackStartMap.get(mapKey))/1000); 
		} else {
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_7";
			desc = String.format("UDPFLOOD伪造源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(mapKey)-attackStartMap.get(mapKey))/1000);
		}
		
		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM", 1l));

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setType("UDPFLOOD");
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamOctets(nUpstreampacket);
		return abnTuple;
	}

}

    